package com.hvgroup.modules.sanic.service.impl;

import com.alibaba.fastjson.JSON;
import com.hvgroup.exception.EntityNotFoundException;
import com.hvgroup.modules.sanic.domain.dto.QaSessionDto;
import com.hvgroup.modules.sanic.domain.enums.DataTypeEnum;
import com.hvgroup.modules.sanic.domain.enums.DiFyAppEnum;
import com.hvgroup.modules.sanic.domain.po.LlmApp;
import com.hvgroup.modules.sanic.domain.po.QaMessage;
import com.hvgroup.modules.sanic.domain.po.QaSession;
import com.hvgroup.modules.sanic.domain.qo.QaQuestionQuery;
import com.hvgroup.modules.sanic.repository.LlmAppRepository;
import com.hvgroup.modules.sanic.repository.QaMessageRepository;
import com.hvgroup.modules.sanic.repository.QaSessionRepository;
import com.hvgroup.modules.sanic.service.DifyQAService;
import com.hvgroup.modules.sanic.service.impl.chat.EventData;
import com.hvgroup.modules.sanic.service.impl.chat.EventStrategy;
import com.hvgroup.modules.sanic.service.impl.chat.chat.EventHandler;
import com.hvgroup.modules.sanic.websocket.QASocketQuery;
import com.hvgroup.modules.sanic.websocket.QAWebSocketHandler;
import com.hvgroup.utils.SecurityUtils;
import com.hvgroup.utils.StringUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
@Service
@RequiredArgsConstructor
public class DifyQAServiceImpl implements DifyQAService {

    private final QaSessionRepository qaSessionRepository;
    private final QaMessageRepository qaMessageRepository;

//    private final ApiTokenRepository apiTokenRepository;
//    private final AppsRepository appsRepository;

    private final LlmAppRepository llmAppRepository;

    private final EventStrategy eventStrategy;

    //    @Value("${dify.api.key.app}")
//    private String difyApiKey;
//
    @Value("${dify.api.url.root}")
    private String difyApiRootUrl;

    @Value("${dify.api.url.chat}")
    private String difyApiChatUrl;

    // SseEmitter
    @Override
    public String wsProcessQuestion(Map<String, Object> body) {
//        String chatId = String.valueOf(body.get("chatId"));
//        String userId = String.valueOf(SecurityUtils.getCurrentUserId());
//        String query = body.get("query").toString();
//        QAWebSocketHandler.sendInfo(QASocketQuery.buildToMap(body));
//, WebSocketSession session
        Map<String, Object> param = new HashMap<>();
        param.put("query", body.get("query"));
        param.put("response_mode", "streaming");
        param.put("user", String.valueOf(SecurityUtils.getCurrentUserId()));
//        param.put("conversation_id",  body.get("conversationId"));

        Map<String, Object> inputs = new HashMap<>();
        inputs.put("qa_type", body.get("qaType"));
        param.put("inputs", inputs);

        // TODO
        String difyApiKey = "";

        QASocketQuery query = QASocketQuery.buildToMap(body, null);
        final WebClient webClient = WebClient.builder().baseUrl(difyApiRootUrl).build();
        // 使用 WebClient 异步调用 DiFy API，订阅事件流
        webClient.post()
                .uri(difyApiChatUrl)
                .header(HttpHeaders.AUTHORIZATION, "Bearer " + difyApiKey)         // Bearer token 前缀已包含在 authToken里
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.TEXT_EVENT_STREAM)                  // 请求 SSE 流
                .bodyValue(param)
                .retrieve()
                .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {
                })
                .subscribe(event -> {
                    String eventType = event.event();
                    String data = event.data();
                    Map<String, Object> d = JSON.parseObject(data, Map.class);
                    try {
                        if (d != null && "MESSAGE".equalsIgnoreCase(String.valueOf(d.get("event")))) {
                            // 解析消息内容（假设 data 即是文本内容或 JSON 字符串）

                            String content = String.valueOf(d.get("answer"));
                            // 过滤 <think> 标签及内容
                            if (StringUtils.isEmpty(content)) {
                                return;
                            }
                            content = content.replaceAll("<think>.*?</think>", "");
                            if (content.trim().isEmpty()) {
                                return; // 如果过滤后为空，不发送给前端
                            }
                            // 发送消息给前端
                            QAWebSocketHandler.sendInfo(QASocketQuery.buildToQuery(query, content));
                        } else if ("MESSAGE_ERROR".equals(eventType)) {
                            // 将错误信息发送给前端或记录日志
                            QAWebSocketHandler.sendInfo(QASocketQuery.buildToQuery(query, ""));
                        }
                    } catch (Exception e) {
                        // 发送消息异常处理（可选日志）
                        log.error(e.getMessage(), e);
                    }
                }, error -> {
                    // 流处理过程中出现异常
                    try {
                        QAWebSocketHandler.sendInfo(QASocketQuery.buildToQuery(query, error.getMessage()));
                    } catch (Exception ex) {
                        log.error(ex.getMessage(), ex);
                        // 忽略进一步错误
                    }
                }, () -> {
                    // 流完整结束后的收尾处理
                    QAWebSocketHandler.sendInfo(QASocketQuery.buildToQuery(query, "end"));
                });

        return "执行完毕";
    }


    /**
     * 请改用通用问答方法 {@link this#fluxProcessQuestion2(QaQuestionQuery)}
     */
    @Deprecated
    @Override
    public Flux<ServerSentEvent<String>> fluxProcessQuestion(QaQuestionQuery body) {

        Map<String, Object> param = body.builderParamBody();
//JSON.toJSONString(param)
        AtomicReference<String> dataType = new AtomicReference<>("");
        AtomicReference<String> busData = new AtomicReference<>("");

        // TODO 2025年6月9日, 删除了直连dify
        return Flux.empty();
//        return Mono.fromCallable(() -> apiTokenRepository.findFirstByAppId(UUID.fromString(body.getAppId()).toString())
//                        .orElseThrow(() -> new RuntimeException("请选择应用!"))
//                ).subscribeOn(Schedulers.boundedElastic())
//                .flatMapMany(apiToken -> {
//                    String difyApiKey = apiToken.getToken();
//                    return WebClient.builder().baseUrl(difyApiRootUrl).build()
//                            .post()
//                            .uri(difyApiChatUrl)
//                            .header(HttpHeaders.AUTHORIZATION, "Bearer " + difyApiKey)
//                            .contentType(MediaType.APPLICATION_JSON)
//                            .accept(MediaType.TEXT_EVENT_STREAM)
//                            .bodyValue(param)
//                            .retrieve()
//                            .bodyToFlux(String.class)
//                            .concatMap(chunk -> {
//
//                                try {
//                                    Map<String, Object> dataJson = JSON.parseObject(chunk);
//
//                                    String eventName = String.valueOf(dataJson.get("event"));
//
//                                    String conversationId = String.valueOf(dataJson.get("conversation_id"));
//                                    String messageId = String.valueOf(dataJson.get("message_id"));
//                                    String taskId = String.valueOf(dataJson.get("task_id"));
//
//
////                                    ObjectMapper objectMapper = new ObjectMapper();
////                                    objectMapper.configure(JsonGenerator.Feature.ESCAPE_NON_ASCII, false);
////                                    JsonNode node = objectMapper.readTree(chunk);
////                                    String readable = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(node);
////                                    System.out.println(readable);
//
//
//                                    if (DiFyCodeEnum.MESSAGE.getCode().equalsIgnoreCase(eventName)) {
//                                        String answer = String.valueOf(dataJson.get("answer"));
//                                        //  普通内容
//                                        if (StringUtils.isNotEmpty(answer) && answer.startsWith("dify_")) {
//                                            String[] event_list = answer.split("_"); // dify 0 t02
//                                            if (event_list.length >= 3 && "0".equals(event_list[1])) {
//                                                // 输出开始
//                                                dataType.set(event_list[2]);
//                                                if (DataTypeEnum.ANSWER.getCode().equalsIgnoreCase(event_list[0])) {
//                                                    return sendMessage(body, answer
//                                                            , "{\"data\": {\"messageType\": \"begin\"}, \"dataType\": \"" + dataType.get() + "\"}"
//                                                            , body.getQaType(), conversationId, messageId, taskId);
//                                                }
//
//                                            } else if (event_list.length >= 3 && "1".equals(event_list[1])) {
//                                                // 输出结束
//                                                dataType.set(event_list[2]);
//                                                if (DataTypeEnum.ANSWER.getCode().equalsIgnoreCase(event_list[0])) {
//                                                    return sendMessage(body, answer
//                                                            , "{\"data\": {\"messageType\": \"end\"}, \"dataType\": \"" + dataType.get() + "\"}"
//                                                            , body.getQaType(), conversationId, messageId, taskId);
//
//
//                                                } else if (StringUtils.isNotEmpty(busData.get()) && DataTypeEnum.BUS_DATA.getCode().equalsIgnoreCase(dataType.get())) {
////                                        res_data = process(json.loads(bus_data)["data"])
//                                                    Map<String, Object> busDataJson = JSON.parseObject(busData.get());
//                                                    String resData = "null";
//                                                    body.setQuery(body.getQuery().split("\\|")[0]);
//
//                                                    if (busDataJson.containsKey("data")) {
//                                                        resData = DbQaDataProcess.process((Map<String, Object>) busDataJson.get("data"));
//                                                    }
//                                                    return sendMessage(body, answer
//                                                            , "{\"data\": " + resData + ", \"dataType\": \"" + dataType.get() + "\"}"
//                                                            , body.getQaType(), conversationId, messageId, taskId);
//                                                }
//                                                dataType.set("");
//                                            }
//                                        } else if (StringUtils.isNotEmpty(dataType.get())) {
//                                            // 这里输出 t02之间的内容
//                                            if (DataTypeEnum.ANSWER.getCode().equalsIgnoreCase(dataType.get())) {
//                                                return sendMessage(body, answer
//                                                        , "{\"data\": {\"messageType\": \"continue\", \"content\": \"" + answer + "\"}, \"dataType\": \"" + dataType.get() + "\"}"
//                                                        , body.getQaType(), conversationId, messageId, taskId);
//                                            }
//
//                                            if (DataTypeEnum.BUS_DATA.getCode().equalsIgnoreCase(dataType.get())) {
//                                                busData.set(answer);
//                                            }
//                                        }
//                                    }
//
//                                    //  异常事件
//                                    if (DiFyCodeEnum.MESSAGE_ERROR.getCode().equalsIgnoreCase(eventName)) {
//                                        String errorMsg = String.valueOf(dataJson.get("message"));
//                                        log.error("调用 dify 失败，错误信息：{}", dataJson);
//
//                                        Map<String, Object> data = new HashMap<>();
//                                        data.put("messageType", "error");
//                                        data.put("content", "调用失败请查看 dify 日志, 错误信息: " + errorMsg);
//
//                                        Map<String, Object> message = new HashMap<>();
//                                        message.put("data", data);
//                                        message.put("dataType", DataTypeEnum.ANSWER.getCode());
//
//                                        return Mono.just(ServerSentEvent.<String>builder()
//                                                .event("error")
//                                                .data(JSON.toJSONString(message))
//                                                .build());
//                                    }
//
//
//                                    return Mono.empty();
//                                } catch (Exception e) {
//                                    log.warn("Error during get_answer: {}", e.getMessage(), e);
//                                    return Mono.just(ServerSentEvent.<String>builder()
//                                            .event("error")
//                                            .data("解析异常:" + e.getMessage())
//                                            .build());
//                                }
//                            })
//                            .onErrorResume(e -> {
//                                log.error("处理 SSE 流时发生错误：{}", e.getMessage(), e);
//                                return Flux.just(ServerSentEvent.<String>builder()
//                                        .event("error")
//                                        .data("流处理异常：" + e.getMessage())
//                                        .build());
//                            })
//                            .concatWith(Mono.just(ServerSentEvent.<String>builder()
//                                    .event("end")
//                                    .data("END")
//                                    .build()));
//
//                });
    }


    /**
     * SSE 格式发送数据，每一行以 data: 开头
     */
    private Mono<ServerSentEvent<String>> sendMessage(QaQuestionQuery body, String answer, String message, String qaType,
                                                      String conversationId, String messageId, String taskId) {

        Map<String, Object> msg = new HashMap<>();
        Map<String, Object> data = new HashMap<>();

        String thinkContent = "";
        if ("dify_1_t04".equals(answer)) {
            msg = JSON.parseObject(message);
            data = (Map<String, Object>) msg.get("data");
            data.put("dataType", DataTypeEnum.BUS_DATA.getCode());
            msg.put("content", data);

        } else if (StringUtils.isNotEmpty(answer) && answer.contains("<think>")) {
            // 处理<think>标签内的内容 (处理deepseek模型思考过程样式)
            Pattern pattern = Pattern.compile("<think>[\\s\\S]*?</think>", Pattern.DOTALL);
            Matcher matcher = pattern.matcher(answer);

            String remainingContent = "";

            if (matcher.find()) {
//                thinkContent = matcher.group(0).replaceAll("<think>|</think>", ""); // 匹配中间内容
                thinkContent = matcher.group(0); // 匹配中间内容
                remainingContent = answer.replaceAll("(?s)<think>[\\s\\S]*?</think>", "").trim(); // (?s) 等同 DOTALL
            } else {
                remainingContent = answer;
            }

            // 发送<think>标签内的内容
            data.put("messageType", "continue");
            data.put("content", "> " + thinkContent.replace("\n", "") + "\n\n" + remainingContent);

            msg.put("data", data);
            msg.put("dataType", DataTypeEnum.ANSWER.getCode());
        } else {
            data.put("messageType", "continue");
            data.put("content", answer);
            msg.put("data", data);
            msg.put("dataType", DataTypeEnum.ANSWER.getCode());
        }

        return saveMessage(msg, thinkContent, body, conversationId, messageId, taskId, qaType);
    }

    /**
     * 保存消息记录并发送SSE数据
     */
    private synchronized Mono<ServerSentEvent<String>> saveMessage(Map<String, Object> message, String think, QaQuestionQuery body, String conversationId, String messageId, String taskId, String qaType) {

        Map<String, Object> data = (Map<String, Object>) message.get("data");

        Mono<Map.Entry<QaSession, QaMessage>> recordMono;

        if (data != null && data.containsKey("content")) {
            recordMono = Mono.fromCallable(() -> addQuestionRecord(
                    conversationId, messageId, taskId,
                    body, think, JSON.toJSONString(message), null, qaType
            ));
        } else if (DataTypeEnum.BUS_DATA.getCode().equalsIgnoreCase(String.valueOf(data.get("dataType")))) {
            recordMono = Mono.fromCallable(() -> addQuestionRecord(
                    conversationId, messageId, taskId,
                    body, think, null, JSON.toJSONString(message), qaType
            )).delayElement(Duration.ofMillis(500));
        } else {
            recordMono = Mono.empty();
        }

        return recordMono
                .subscribeOn(Schedulers.boundedElastic())
                .map(result -> {
                    message.put("chatId", result.getKey().getChatId());
                    message.put("conversationId", result.getKey().getConversationId());
                    message.put("feadId", result.getValue().getId());
//                    message.pu
                    return JSON.toJSONString(message);
                })
                .defaultIfEmpty(JSON.toJSONString(message)) // 如果 recordMono 是空的，使用原始 message
                .map(modifiedMessage ->
                        ServerSentEvent.<String>builder()
                                .event("message")
                                .data(modifiedMessage)
                                .build()
                );
    }

    /**
     * 记录用户问答记录，如果记录已存在，则更新之；否则，创建新记录。
     */
    @Transactional
    public synchronized Map.Entry<QaSession, QaMessage> addQuestionRecord(String conversationId, String messageId, String taskId, QaQuestionQuery query, String think, String to2Answer, String to4Answer, String qaType) {

        String fileKey = "";
        String question = query.getQuery();
        if (DiFyAppEnum.FILEDATA_QA.getCode().equalsIgnoreCase(qaType)) {
            fileKey = question.split("\\|")[0];
            if (question.contains("|")) {
                question = question.split("\\|")[1];
            } else {
                question = null;
            }
        }
        // 查询t_user_qa_record
//        sql = f"select * from t_user_qa_record where user_id={user_id} and chat_id='{chat_id}' and message_id='{message_id}'"
        QaSession qaSession;

        if (query.getChatId() == null || StringUtils.isEmpty(conversationId)) {
            // 新会话
            qaSession = new QaSession();
            qaSession.setConversationId(conversationId);
            qaSession.setTitle(question);
            qaSession.setEnable(true);
            qaSession.setUserId(query.getUserId());
            qaSession.setAppId(query.getAppId());
//            appsRepository.findById(UUID.fromString(query.getAppId()).toString()).ifPresent(app -> {
//                qaSession.setAppName(app.getName());
//            });
            qaSessionRepository.save(qaSession);
            query.setChatId(qaSession.getChatId());
        } else {
            // 已有会话
            qaSession = qaSessionRepository.findById(query.getChatId())
                    .orElseThrow(() -> new EntityNotFoundException(QaSession.class, "", ""));
            // 更新时间
            qaSession.setUpdateTime(new Timestamp(System.currentTimeMillis()));
            qaSessionRepository.save(qaSession);
        }

        // 创建消息记录
        QaMessage qaMessage = new QaMessage();
        qaMessage.setChatId(qaSession.getChatId());
        qaMessage.setQuestion(question);
        qaMessage.setQaType(qaType);
        qaMessage.setMessageId(messageId);
        qaMessage.setTaskId(taskId);
        qaMessage.setFileKey(fileKey);
        qaMessage.setFileUrl(query.getFileUrl());
        qaMessage.setFileSize(query.getFileSize());
        qaMessage.setModelThink(StringUtils.defaultString(think, "").replaceAll("<think>|</think>", ""));
        qaMessage.setModelAnswer(StringUtils.defaultString(to2Answer, to4Answer));
        qaMessage.setRating(0);
        qaMessageRepository.save(qaMessage);

        return new AbstractMap.SimpleEntry<>(qaSession, qaMessage);
    }

    @Override
    public Mono<String> fluxProcessSuggested(QaQuestionQuery body) {
        // TODO 2025年6月9日, 删除了直连dify
//
//        // 并行执行两个阻塞操作，分别查 message 和 apiToken
//        Mono<QaMessage> messageMono = Mono.fromCallable(() ->
//                qaMessageRepository.findFirstByChatIdOrderByCreateTimeDesc(body.getChatId())
//                        .orElseThrow(() -> new IllegalArgumentException("找不到 message_id 对应的记录"))
//        ).subscribeOn(Schedulers.boundedElastic());
//
//
//        Mono<ApiToken> tokenMono = Mono.fromCallable(() ->
//                apiTokenRepository.findFirstByAppId(UUID.fromString(body.getAppId()).toString())
//                        .orElseThrow(() -> new RuntimeException("请选择应用!"))
//        ).subscribeOn(Schedulers.boundedElastic());
//
//        // 将两个结果组合处理
//        return Mono.zip(messageMono, tokenMono)
//                .flatMap(tuple -> {
//                    QaMessage session = tuple.getT1();
//                    ApiToken apiToken = tuple.getT2();
//
//                    String messageId = session.getMessageId();
//                    String difyApiKey = apiToken.getToken();
//                    String path = DiFyRestApi.DIFY_REST_SUGGESTED.replace(":message_id", messageId);
//
//                    log.info("调用 Dify: GET {}/{}?user={}", difyApiRootUrl, path, body.getUser());
//
//                    return WebClient.builder()
//                            .baseUrl(difyApiRootUrl)
//                            .build()
//                            .get()
//                            .uri(uriBuilder -> uriBuilder
//                                    .path(path)
//                                    .queryParam("user", body.getUser())
//                                    .build()
//                            )
//                            .header(HttpHeaders.AUTHORIZATION, "Bearer " + difyApiKey)
//                            .accept(MediaType.APPLICATION_JSON)
//                            .retrieve()
//                            .onStatus(HttpStatus::isError, resp ->
//                                    resp.bodyToMono(String.class)
//                                            .flatMap(bodyStr -> {
//                                                log.error("Dify 错误响应 [{}]: {}", resp.statusCode(), bodyStr);
//                                                return Mono.error(new RuntimeException("调用 Dify 失败: " + bodyStr));
//                                            })
//                            )
//                            .bodyToMono(String.class);
//                })
//                .onErrorResume(e -> {
//                    log.error("调用失败: {}", e.getMessage(), e);
//                    String jsonError = "{\"error\": \"" + e.getMessage().replace("\"", "'") + "\"}";
//                    return Mono.just(jsonError);
//                });

        return Mono.empty();

//
//        return Mono.fromCallable(() ->
//                        qaMessageRepository.findFirstByChatIdOrderByCreateTimeDesc(body.getChatId())
//                                .orElseThrow(() -> new IllegalArgumentException("找不到 message_id 对应的记录"))
//                )
//                .subscribeOn(Schedulers.boundedElastic())
//                .flatMap(session -> {
//                    String messageId = session.getMessageId();
//                    String path = DiFyRestApi.DIFY_REST_SUGGESTED.replace(":message_id", messageId);
//
//                    log.info("调用 Dify: GET {}/{}?user={}", difyApiUrl, path, body.getUser());
//
//                    return WebClient.builder()
//                            .baseUrl(difyApiUrl)
//                            .build()
//                            .get()
//                            .uri(
//                                    uriBuilder -> uriBuilder
//                                            .path(path)
//                                            .queryParam("user", body.getUser())
//                                            .build()
//                            )
//                            .header(HttpHeaders.AUTHORIZATION, "Bearer " + difyApiKey)
//                            .accept(MediaType.APPLICATION_JSON)
//                            .retrieve()
//                            .onStatus(HttpStatus::isError, resp ->
//                                    resp.bodyToMono(String.class)
//                                            .flatMap(bodyStr -> {
//                                                log.error("Dify 错误响应 [{}]: {}", resp.statusCode(), bodyStr);
//                                                return Mono.error(new RuntimeException("调用 Dify 失败: " + bodyStr));
//                                            })
//                            )
//                            .bodyToMono(String.class);
//                }).onErrorResume(e -> {
//                    log.error("调用失败: {}", e.getMessage(), e);
//                    String jsonError = "{\"error\": \"" + e.getMessage().replace("\"", "'") + "\"}";
//                    return Mono.just(jsonError);
//                });

    }


    // --------------------------------------------------------------------------------------------

    /**
     * 核心业务流：SSE 流式问答全链路
     */
    @Override
    public Flux<ServerSentEvent<String>> fluxProcessQuestion2(QaQuestionQuery body) {
        // 1. 异步获取 API Token，防止阻塞主线程
        Mono<LlmApp> tokenMono = Mono.fromCallable(() -> llmAppRepository.findFirstByAppId(body.getAppId())
                                .orElseThrow(() -> new RuntimeException("请选择应用!"))
        ).subscribeOn(Schedulers.boundedElastic());

        // 2. 拼接流式 SSE 响应
        return tokenMono.flatMapMany(apiToken -> {
            String difyApiKey = apiToken.getAccessToken();
            WebClient webClient = WebClient.builder()
                    .baseUrl(difyApiRootUrl)
                    .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + difyApiKey)
                    .build();
            Map<String, Object> param = body.builderParamBody();

            // TODO 需要调整为数据库获取接口url
            // 2.1 主体事件流：scan 递进处理业务事件
            Flux<EventData> eventDataFlux = webClient.post()
                    .uri(difyApiChatUrl)
                    .contentType(MediaType.APPLICATION_JSON)
                    .accept(MediaType.APPLICATION_JSON) // 用 JSON，避免\n分割问题
                    .bodyValue(param)
                    .retrieve()
                    .bodyToFlux(String.class)
                    .scan(EventData.builder().busData(body).requestTimestamp(System.currentTimeMillis()).build(),
                            (eventData, chunk) -> {
                        Map<String, Object> map = JSON.parseObject(chunk, Map.class);
                        // 遍历所有 handler，依次判定与处理
                        for (EventHandler handler : eventStrategy.getHandlers(apiToken.getMode())) {
                            if (handler.supports(map)) {
                                eventData = handler.handle(map, eventData);
                            }
                        }
                        return eventData;
                    });

            // 2.2 只推送业务有效变化事件
            return eventDataFlux
                    .skip(1) // 跳过初始状态
                    .distinctUntilChanged(EventData::getStep) // 只推送 step 变更
                    .filter(o -> o.getStep() > 0) // 有效业务
                    .delayElements(Duration.ofMillis(500))
                    // .filter(e -> StringUtils.isNotBlank(e.getText())) // 如需再判空可加
                    .map(eventData -> {
                        String data = Optional.ofNullable(eventData)
                                .orElseGet(() -> EventData.builder().text("").build())
                                .getText();
                        return ServerSentEvent.<String>builder()
                                .event("message")
                                .data(data)
                                .build();
                    })
                    .onErrorResume(e -> {
                        log.error("处理 SSE 流时发生错误：{}", e.getMessage(), e);
                        return Flux.just(ServerSentEvent.<String>builder()
                                .event("error")
                                .data("流处理异常：" + e.getMessage())
                                .build());
                    })
                    .concatWith(Mono.just(ServerSentEvent.<String>builder()
                            .event("end")
                            .data("END")
                            .build()));
        });
    }


    @Override
    public void ratingAnswer(Map<String, Object> body) {
        Long id = Long.valueOf(StringUtils.defaultIfEmpty(String.valueOf(body.get("id")), "-1"));
        Integer rating = Integer.valueOf(StringUtils.defaultIfEmpty(String.valueOf(body.get("rating")), "1"));
        QaMessage qaMessage = qaMessageRepository.findById(id).orElseThrow(() -> new EntityNotFoundException(QaMessage.class, "", ""));
        qaMessage.setRating(rating);
        qaMessageRepository.save(qaMessage);
    }

    @Override
    public void deleteMessage(Map<String, Object> body) {
        List<Long> ids = JSON.parseArray(StringUtils.defaultIfEmpty((String) body.get("ids"), "[]"), Long.class);
        qaMessageRepository.deleteAllById(ids);
    }

    @Transactional
    @Override
    public void deleteSession(Map<String, Object> body) {
        Long chatId = Long.valueOf(StringUtils.defaultIfEmpty(String.valueOf(body.get("chatId")), "-1"));
        qaMessageRepository.deleteByChatId(chatId);
        qaSessionRepository.deleteById(chatId);
    }

    @Override
    public void updateSession(QaSessionDto body) {
        QaSession qaSession = qaSessionRepository.findById(body.getChatId())
                .orElseThrow(() -> new EntityNotFoundException(QaSession.class, "", ""));
        qaSession.setTitle(body.getTitle());
        qaSessionRepository.save(qaSession);
    }
}

